package com.need5

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Row

/**
  * Created by zhuang on 2018/3/5.
  */
object DealDataSec {
  //处理appid对应appname
  //这里没有实现自定义特质，应该实现特质，重写特质的具体指定方法实现分工合作，特质的参数会设置成如下
  //方便随意传参数
  def id2Name(args: Any*): Map[String, String] = {
    var rdd: RDD[String] = args(0).asInstanceOf[RDD[String]]
    rdd.map(_.split("\t")).filter(_.length >= 5).map(t => (t(4), t(1))).collect.toMap
  }

  //根据id查找appname
  def dealAppname(args: Any*): String = {
    var id: String = args(0).asInstanceOf[String]
    var name: String = args(1).asInstanceOf[String]
    var bc: Broadcast[Map[String, String]] = args(2).asInstanceOf[Broadcast[Map[String, String]]]

    val mp = bc.value
    if (name.isEmpty) {
      if (!id.isEmpty) {
        //因为这个文件里面的appid也有可能是空的
        mp.getOrElse(id, id)
      } else {
        "未知"
      }
    } else {
      name
    }
  }

  //处理第一个字段，（广告位类型）
  def adType(i: Int): String = {
    if (i < 10) s"LC0$i" else s"LC$i"
  }

  //处理操作系统
  def cli(i: Int): String = {
    i match {
      case 1 => "D00010001"
      case 2 => "D00010002"
      case 3 => "D00010003"
      case _ => "D00010004"
    }
  }

  //联网方式
  def dealNet(args:Any*): String = {
    var n = args(0).asInstanceOf[String]
    n match {
      case "WIFI" => "D00020001"
      case "4G" => "D00020002"
      case "3G" => "D00020003"
      case "2G" => "D00020004"
      case _ => "D00020005"
    }
  }

  //运营商名称
  def dealIsp(i: String): String = {
    i match {
      case "移动" => "D00030001"
      case "联通" => "D00030002"
      case "电信" => "D00030003"
      case _ => "D00030004"
    }
  }

  //利用diff方法去除求差集处理关键字
  def dealKeyW(key: String, bc: Broadcast[Array[String]]): String = {
    val split = key.split("[|]")
    //去除敏感词
    val diff: Array[String] = split.diff(bc.value)
    val toBuffer = diff.toBuffer
    for (li <- diff) {
      if (li.length > 8 || li.length < 3) {
        toBuffer -= li
      }
    }
    toBuffer.mkString(",")
  }

  //处理关键字的方法利用map的方法
  /*def dealKeyW(key: String, bcMap: Broadcast[Map[String, String]]): String = {
    //if (key.contains("[|]")) {
    val split = key.split("[|]")
    var toList: List[String] = split.toList
    val toBuffer: mutable.Buffer[String] = toList.toBuffer
    val value: Map[String, String] = bcMap.value
    for (li <- toList) {
      if (li.length > 8 || li.length < 3 || !value.get(li).isEmpty) {
        toBuffer -= li
      }
    }
    toBuffer.mkString(",")
}*/

  //处理十五条数据，返回第一个不是空的即可
  def dealData_15(arr: Array[String]): String = {
    var flag = true
    var res: String = ""
    //一旦有符合的直接停止循环返回值即可
    for (i <- arr if flag) {
      if (i.length != 0) {
        flag = false
        res = i
      }
    }
    res
  }

  //拿字段
  def getField(row: Row, bc: Broadcast[Map[String, String]], bcMap: Broadcast[Array[String]]) = {
    val imei = row.getAs[String]("imei")
    val mac = row.getAs[String]("mac")
    val idfa = row.getAs[String]("idfa")
    val openudid = row.getAs[String]("openudid")
    val androidid = row.getAs[String]("androidid")
    val imeimd5 = row.getAs[String]("imeimd5")
    val macmd5 = row.getAs[String]("macmd5")
    val idfamd5 = row.getAs[String]("idfamd5")
    val openudidmd5 = row.getAs[String]("openudidmd5")
    val androididmd5 = row.getAs[String]("androididmd5")
    val imeisha1 = row.getAs[String]("imeisha1")
    val macsha1 = row.getAs[String]("macsha1")
    val idfasha1 = row.getAs[String]("idfasha1")
    val openudidsha1 = row.getAs[String]("openudidsha1")
    val androididsha1 = row.getAs[String]("androididsha1")
    var arr = Array[String](
      imei,
      mac,
      idfa,
      openudid,
      androidid,
      imeimd5,
      macmd5,
      idfamd5,
      openudidmd5,
      androididmd5,
      imeisha1,
      macsha1,
      idfasha1,
      openudidsha1,
      androididsha1
    )

    //开始处理这十五条数据，返回第一个不是空的即可
    val userid = dealData_15(arr)
    //开始拿字段
    //第一个字段
    val adspacetype = row.getAs[Int]("adspacetype")
    //处理广告类型字段
    val adTyped = adType(adspacetype)
    //第二个字段
    val appName = row.getAs[String]("appname")
    val appid = row.getAs[String]("appid")
    //app名称开始处理并返回appnamed
    val appnamed = dealAppname(appid, appName, bc)
    //第三个字段
    val aderid = row.getAs[Int]("adplatformproviderid")
    //处理
    val aderided = s"CN$aderid"
    //第四个字段
    //操作系统
    val client = row.getAs[Int]("client")
    //处理
    val cliented = cli(client)
    //联网方式
    val networkmanner = row.getAs[String]("networkmannername")
    //处理
    val networkmannered = dealNet(networkmanner)
    //运营商名称
    val ispname = row.getAs[String]("ispname")
    //处理
    val ispnamed = dealIsp(ispname)
    //第五个字段
    val keyword = row.getAs[String]("keywords")
    //处理关键字
    val keyWorded: String = dealKeyW(keyword, bcMap)
    //第六个关键字cityname
    val proName = row.getAs[String]("provincename")
    //处理
    val proNamed = s"ZP$proName"
    // 第七个关键字cityname
    val cName = row.getAs[String]("cityname")
    //处理
    val cNamed = s"ZC$cName"

    //返回值
    (userid, List[String](adTyped, appnamed, aderided, cliented, networkmannered, ispnamed, keyWorded, proNamed, cNamed))

  }
}
